package com.heima.kafka.simple;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

/**
 * 生产商开始
 *
 * @author DiXiaoNie
 * @date 2022/12/28
 * @since 2022/12/28 10:42
 */

public class ProducerStart {
    public static void main(String[] args) {
        System.out.println("start~~~~~~~~~~~~~~~~~~~~~~~");

        //2.设置kafka配置信息
        Properties properties = new Properties();
        //2.1 kafka地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.56.130:9092");
        //2.2 设置对key value进行字符串序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        //重试配置
        properties.put(ProducerConfig.RETRIES_CONFIG, 10);
        //1.创建生产者对象
        KafkaProducer kafkaProducer = new KafkaProducer(properties);


        for (int i = 1; i <= 5; i++) {
            if (i % 2 == 0) {
                ProducerRecord record = new ProducerRecord("kafka-stream-producer", "key" + i, "hello kafka");
                kafkaProducer.send(record);
            } else {
                ProducerRecord record = new ProducerRecord("kafka-stream-producer", "key" + i, "hello itcast");
                kafkaProducer.send(record);
            }

        }


        /*        //3.发送消息
         *//**
         *参数1 表示消息队列名称
         *参数2 表示消息得key
         *参数3 表示的是消息的内容
         * *//*
        ProducerRecord producerRecord = new ProducerRecord("kafka-stream-producer", "key_start", "hello kafka");
*//*        Future future = kafkaProducer.send(producerRecord);
        try {
            RecordMetadata recordMetadata = (RecordMetadata) future.get();
            System.out.println(recordMetadata.offset());
        } catch (Exception e) {
            e.printStackTrace();
        }*//*
        //异步发送
        kafkaProducer.send(producerRecord, new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                if (e == null) {
                    System.out.println(recordMetadata.offset());
                } else {
                    System.out.println(e);
                }
            }
        });
        //4.关闭流*/
        kafkaProducer.close();
        System.out.println("end~~~~~~~~~~~~~~~~~");

    }
}
